package com.roncoo.eshop.storm;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

import com.roncoo.eshop.storm.bolt.LogParseBolt;
import com.roncoo.eshop.storm.bolt.ProductCountBolt;
import com.roncoo.eshop.storm.spout.AccessLogKafkaSpout;

/**
 * Copyright: Copyright (c) 2019 Hu-Cheng
 *
 * @ClassName: HotProductTopology.java
 * @Description: storm实时计算热门商品，热数据统计拓扑 
 * 				 项目描述：
 * 					   1.AccessLogKafkaSpout从kafka获取数据，传递给bolt
 * 					   2.LogParseBolt将数据解析，获取productId，在传递给下一个bolt
 * 					   3.ProductCountBolt将productId数据使用LRU计算,计算出热门商品数据，并将数据通过taskid进行分段存储，保存到zookeeper节点中
 *
 * @version: v1.0.0
 * @author: hucheng
 * @date: Oct 15, 2019 2:02:05 PM
 *
 */
public class HotProductTopology {
	public static void main(String[] args) {
		TopologyBuilder builder = new TopologyBuilder();

		builder.setSpout("AccessLogKafkaSpout", new AccessLogKafkaSpout(), 1);
		builder.setBolt("LogParsrBolt", new LogParseBolt(), 2).setNumTasks(2).shuffleGrouping("AccessLogKafkaSpout");
		builder.setBolt("ProductCountBolt", new ProductCountBolt(), 2).setNumTasks(2).fieldsGrouping("LogParsrBolt",
				new Fields("productId"));

		Config config = new Config();

		// 服务器启动
		if (args != null && args.length > 0) {
			config.setNumWorkers(3);
			try {
				StormSubmitter.submitTopology(args[0], config, builder.createTopology());
			} catch (Exception e) {
				e.printStackTrace();
			}
		} else {
			// 说明是在eclipse里面本地运行
			config.setMaxTaskParallelism(20);
			LocalCluster cluster = new LocalCluster();
			cluster.submitTopology("HotProductTopology", config, builder.createTopology());
//			Utils.sleep(30000);
//			cluster.shutdown();
		} 
	}
}
